package com.example.sys.async.service;

import com.baomidou.mybatisplus.core.metadata.TableInfo;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import com.example.core.cluster.MasterServerService;
import com.example.core.debezium.listener.SysDebeziumDataChangeListener;
import com.example.core.debezium.message.Message;
import com.example.core.mapping.Mapping;
import com.example.core.service.BaseServiceImpl;
import com.example.sys.async.dao.SysASyncDao;
import com.example.sys.async.mapping.SysASyncMapping;
import com.example.sys.async.po.SysASync;
import com.example.sys.async.task.RunTask;
import com.example.sys.async.vo.SysASyncVo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.*;

@Service
public class SysASyncServiceImpl extends BaseServiceImpl<SysASyncDao, SysASync, SysASyncVo> implements SysASyncService, SysDebeziumDataChangeListener, MasterServerService {

    Logger logger = LoggerFactory.getLogger(getClass());

    private Thread thread;

    private volatile boolean isStop = false;

    private Object object = new Object();

    private ThreadPoolExecutor threadPool = null;

    @Override
    public Class<? extends Mapping> getMappingClass() {
        return SysASyncMapping.class;
    }

    @Override
    public void changeEvent(Message message) {
        synchronized (object) {
            TableInfo tableInfo = TableInfoHelper.getTableInfo(SysASync.class);
            if (message.getHandleType().equalsIgnoreCase(Message.CREATE) && message.getTable().equalsIgnoreCase(tableInfo.getTableName())) {
                object.notify();
            }
        }
    }

    @Override
    public void start() {
        this.threadPool = new ThreadPoolExecutor(
                10,
                20,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(1000)
        );
        this.thread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (!isStop) {
                    synchronized (object) {
                        List<SysASync> list = findList();
                        if (list.isEmpty()) {
                            try {
                                object.wait();
                            } catch (Exception e) {
                                logger.warn("线程暂停报错: " + e.getMessage());
                            }
                        } else {
                            try {
                                CountDownLatch countDownLatch = new CountDownLatch(list.size());
                                for (SysASync sysASync : list) {
                                    threadPool.execute(new RunTask(countDownLatch, sysASync, getApplicationContext()));
                                }
                                countDownLatch.await();
                            } catch (Exception e) {
                                logger.warn("执行任务报错: " + e.getMessage());
                            }
                        }
                    }
                }
            }
        });
        this.thread.start();
    }

    @Override
    public void stop() {
        this.isStop = true;
        if (this.thread.isAlive()) {
            this.thread.interrupt();
        }
        if (!this.threadPool.isShutdown()) {
            this.threadPool.isTerminated();
        }
    }

    /***
     * 查询所有待执行的任务数据
     * @return
     */
    @Override
    public List<SysASync> findList() {
        return getBaseMapper().findList();
    }
}
